package com.atguigu.flink.sql.connector;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by 黄凯 on 2023/6/25 0025 18:07
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 */
public class Flink01_DataGenConnector {

    public static void main(String[] args) throws Exception {

        // 1. 创建表环境
        TableEnvironment tableEnv =
                TableEnvironment.create(EnvironmentSettings.newInstance().inStreamingMode().build());

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv1 = StreamTableEnvironment.create(env);

        // 2. 创建DataGen连接器表
        String dataGenSql = "create table t1(\n" +
                "\n" +
                "    id string,\n" +
                "    vc int,\n" +
                "    ts bigint\n" +
                "\n" +
                ")with(\n" +
                "\n" +
                "    'connector' = 'datagen',\n" +
                "    'rows-per-second' = '1' ,\n" +
                "    'fields.id.kind' = 'random' ,\n" +
                "    'fields.id.length' = '5' ,\n" +
                "    'fields.vc.kind' = 'random' ,\n" +
                "    'fields.ts.kind' = 'sequence' ,\n" +
                "    'fields.ts.start' = '1000' ,\n" +
                "    'fields.ts.end' = '10000'\n" +
                "\n" +
                ")";

        tableEnv1.executeSql(dataGenSql);

        //3. 表的查询转换
//        tableEnv1.sqlQuery("select * from t1 ").execute().print();

        //4. 写出数据  print Connector
        String printSql =
                " create table t2 (" +
                        " id STRING , " +
                        " vc INT , " +
                        " ts BIGINT " +
                        ") WITH ( " +
                        " 'connector' = 'print' "+
                        ")" ;

        tableEnv1.executeSql(printSql);

        //从t1表查询数据，写入到t2
        tableEnv1.executeSql("insert into t2 select * from t1 ");

        env.execute();


    }

}
